package com.wei.mutualfriend;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;

/**
 *  A:B,C,D,F,E,O
	B:A,C,E,K
	C:F,A,D,I
	D:A,E,F,L
	
	A-B:C A-B:D
	
 * @author Administrator
 *
 */
public class MutualFriend {
	
	static class MutualFriendMap extends Mapper<LongWritable, Text, Text, Text>{
		Text masterText=new Text();
		Text friendText=new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String[] split = value.toString().split(":");
			String master=split[0];
			masterText.set(master);
			String friendStr=split[1];
			String[] friends = friendStr.split(",");
			for (String friend : friends) {
				friendText.set(friend);
				context.write(friendText, masterText);
			}
			
		}
	}
	
	static class MutualFriendReduce extends Reducer<Text, Text, Text, Text>{
		Text text=new Text();
		@Override
		protected void reduce(Text key, Iterable<Text> iterable,Context context)
				throws IOException, InterruptedException {
			StringBuffer buffer=new StringBuffer();
			for (Text text : iterable) {
				buffer.append(text.toString()).append(",");
			}
			text.set(buffer.toString());
			context.write(key, text);
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(MutualFriend.class);

		job.setMapperClass(MutualFriendMap.class);
		job.setReducerClass(MutualFriendReduce.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}

}
